"""
图存储
1、存边
A->B
A->D
B->C
C->A
C->B
D->C
D->B

2、存顶点
A,B|D
B,C
C,A|B
D,B|C
"""
from doctest import master

from pyspark.context import SparkContext

sc = SparkContext(master="local")

# 1、读取数据
lines_rdd = sc.textFile("../../data/pages.txt")


# 2、整理数据
def map_fun(line):
    split = line.split(",")
    page = split[0]
    linked = split[1].split("|")
    return page, linked


pages_rdd = lines_rdd.map(map_fun)

# 多次使用
pages_rdd.cache()

# 统计网页数量
N = pages_rdd.count()

# 2、给每个网页设置初始PR值
init_pr_rdd = pages_rdd.map(lambda kv: (kv[0], kv[1], 1.0))

# 循环迭代重复计算新的PR值
while True:
    # 3、将每个网页的PR平分给出链列表
    def avg_pr_fun(kv):
        linked = kv[1]
        pr = kv[2]
        # 计算该网页分给其他网页的PR值
        avg_pr = pr / len(linked)

        # 将分到PR值放到出链列表中
        linked_pr = [(link, avg_pr) for link in linked]

        return linked_pr


    avg_pr_rdd = init_pr_rdd.flatMap(avg_pr_fun)

    # 4、统计每个网页分到的PR值
    new_pr_rdd = avg_pr_rdd.reduceByKey(lambda a, b: a + b)

    # 计算收敛
    # 一、计算当前PR值和上一次PR值的差值平均值
    sum_diff = init_pr_rdd \
        .map(lambda kv: (kv[0], kv[-1])) \
        .join(new_pr_rdd) \
        .map(lambda kv: abs(kv[1][0] - kv[1][1])) \
        .sum()

    avg_diff = sum_diff / N
    print(avg_diff)

    # 5、关联出链列表
    new_page_rdd = new_pr_rdd \
        .join(pages_rdd) \
        .map(lambda kv: (kv[0], kv[1][1], kv[1][0]))

    # 二、收敛
    if avg_diff < 0.1:
        # 保存结果
        new_page_rdd.saveAsTextFile("../../data/pagerank")
        break

    new_page_rdd.foreach(print)

    # 6、切换RDD进入下一次迭代
    init_pr_rdd = new_page_rdd

